package com.atguigu.api3

import com.atguigu.api.SensorReading
import org.apache.flink.api.common.functions.RichFlatMapFunction
import org.apache.flink.api.common.state._
import org.apache.flink.streaming.api.scala._
import org.apache.flink.util.Collector

/**
 *
 * @description: 在processTime处理模式下,对延时数据可以直接将时间延时若干单位时间,
 *               在eventTime处理模式下,对延时数据就可以通过watermark机制来处理,watermark用于表示timestamp小于Watermark的数据都已经达到.用来让程序自己平衡延迟和结果正确性
 *               watermark是一条特殊的数据记录
 *               必须单调递增,以确保任务的事件时间时钟都在向前推进,而不是在后退,watermark与数据的时间戳相关,
 *               首先确定延时3s[数据中最大差值],0-5s的窗口 6-10 首先数据都各自放到相应
 *               的桶里面,然后来了8 8-3=5那么0-5这个窗口就可以关闭计算了,后来的4则丢弃,有了watermark后延时处理数据关闭窗口也是以watermark为基准的.如210-225 225-245 来了一个285s的数据那么当前watermark为285-3=282,
 *               此时282距离225s为57s还达不到关窗的条件,如果来到了289s数据进来了.那么watermark=286 286-225=61延时也满足了直接关窗聚合计算了.
 * @time: 2021-03-15 15:35
 * @author: baojinlong 
 **/
object F03StateTestDemo {
  def main(args: Array[String]): Unit = {
    val environment: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    // 设置并行度
    environment.setParallelism(1)
    // 设置状态后端
    val inputStream: DataStream[String] = environment.socketTextStream("localhost", 7777)
    val dataStream: DataStream[SensorReading] = inputStream.map {
      data => {
        val arr: Array[String] = data.split(",")
        SensorReading(arr(0), timestamp = arr(1).toLong, temperature = arr(2).toDouble)
      }
    }

    // 需求:对于温度传感器温度跳变值超过10摄氏度就报警 ctrl alt shift +/-可以查看隐式转换 方式一
    //    val resultDataStream: DataStream[(String, Double, Double)] = dataStream
    //      .keyBy(_.id)
    //      .flatMap(new MyDiyFlatMapClass(10.0))

    val resultDataStream: DataStream[(String, Double, Double)] = dataStream
      .keyBy(_.id)
      // (T, Option[S]) => (TraversableOnce[R], Option[S])
      .flatMapWithState[(String, Double, Double), Double] {
        case (data: SensorReading, None) => (List.empty, Some(data.temperature))
        case (data: SensorReading, lastTemp: Some[Double]) =>
          val diff: Double = (data.temperature - lastTemp.get).abs
          if (diff > 10.0) {
            (List((data.id, lastTemp.get, data.temperature)), Some(data.temperature))
          } else {
            (List.empty, Some(data.temperature))
          }

      }
    resultDataStream.print("all data")
    environment.execute("state test")
  }

}

class MyDiyFlatMapClass(threshold: Double) extends RichFlatMapFunction[SensorReading, (String, Double, Double)] {
  // 定义状态保存上一次的温度值
  lazy val lastTemperatureValue: ValueState[Double] = getRuntimeContext.getState(new ValueStateDescriptor[Double]("lastTemperatureValue", classOf[Double]))

  /**
   * 注意flatMap收集器用collect
   *
   * @param in
   * @param collector
   */
  override def flatMap(in: SensorReading, collector: Collector[(String, Double, Double)]): Unit = {
    // 获取上次的温度值
    val oldValue: Double = lastTemperatureValue.value
    // 获取最新的温度值求差值比较
    val diffValue: Double = (in.temperature - oldValue).abs
    if (diffValue > threshold) {
      // 符合条件的把数据往外输出
      collector.collect((in.id, oldValue, in.temperature))
    }
    // 更新状态
    lastTemperatureValue.update(in.temperature)
  }
}

